package com.company.project.rocketmq.processor;

import javax.annotation.Resource;

import org.apache.rocketmq.common.message.MessageClientExt;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import tk.mybatis.mapper.entity.Condition;

import com.alibaba.fastjson.JSON;
import com.company.project.constant.MQEnums;
import com.company.project.model.TradeMqConsumerLog;
import com.company.project.protocol.goods.AddGoodsNumberReq;
import com.company.project.protocol.mq.CancelOrderMQ;
import com.company.project.rocketmq.IMessageProcessor;
import com.company.project.service.TradeGoodsService;
import com.company.project.service.TradeMqConsumerLogService;

/**
 * 商品库存 消息去重，保证幂等性（这里采用记录表的方式）
 *
 */
@Component
public class GoodsCancelOrderProcessor implements IMessageProcessor {
	private static final Logger LOGGER = LoggerFactory.getLogger(GoodsCancelOrderProcessor.class);
	@Resource
	private TradeGoodsService tradeGoodsService;

	/* 消费日志表（trade-mq-consumer-log） */
	@Autowired
	private TradeMqConsumerLogService tradeMqConsumerLogMapper;

	public boolean handleMessage(MessageExt messageExt) {
		TradeMqConsumerLog mqConsumerLog = null;
		try {
			String groupName = "goods_ace_trade_cancel_group";
			String body = new String(messageExt.getBody(), "UTF-8");
			String msgId = messageExt.getMsgId();
			// rockermq 具体用的哪个 msgId ? TODO
			if (messageExt instanceof MessageClientExt) {
				msgId = ((MessageClientExt) messageExt).getOffsetMsgId();
			}
			String tags = messageExt.getTags();
			String keys = messageExt.getKeys();
			LOGGER.info("GoodsCancelOrderProcessor receive message:" + messageExt);
			TradeMqConsumerLog queryLog = new TradeMqConsumerLog();
			// 联合主键groupName以及tag，
			queryLog.setGroupName(groupName);
			queryLog.setMsgKeys(keys);
			queryLog.setMsgTag(tags);
			// 从MQ，消费日志表
			mqConsumerLog = this.tradeMqConsumerLogMapper.selectOne(queryLog);
			if (mqConsumerLog != null) {
				String consumerStatus = mqConsumerLog.getConsumerStatus();
				// 消息已经处理过了
				if (MQEnums.ConsumerStatusEnum.SUCCESS.getStatusCode().equals(consumerStatus)) {
					LOGGER.warn("已经处理过了，不能再处理了");
					return true;
				} else if (MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode().equals(consumerStatus)) {
					LOGGER.warn("正在处理稍后再试");
					return false;
				} else if (MQEnums.ConsumerStatusEnum.FAIL.getStatusCode().equals(consumerStatus)) {
					if (mqConsumerLog.getConsumerTimes() >= 3) {
						// 让这个消息不再重试，转人工处理
						LOGGER.warn("超过三次，不再处理");
						return true;
					}
					TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
					updateMqConsumerLog.setGroupName(groupName);
					updateMqConsumerLog.setMsgKeys(keys);
					updateMqConsumerLog.setMsgTag(tags);
					updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode());
					// 乐观锁的方式防止并发更新，只有一个处理者能成功
					Condition c = new Condition(TradeMqConsumerLog.class);
					c.createCriteria().andEqualTo("msgKeys", mqConsumerLog.getMsgKeys());
					c.createCriteria().andEqualTo("groupName", mqConsumerLog.getGroupName());
					c.createCriteria().andEqualTo("msgTag", mqConsumerLog.getMsgTag());
					c.createCriteria().andEqualTo("consumerTimes", mqConsumerLog.getConsumerTimes());
					int record = this.tradeMqConsumerLogMapper.updateByConditionSelective(updateMqConsumerLog, c);
					if (record <= 0) {
						LOGGER.warn("并发更新处理状态，一会儿重试");
						return false;
					}

				}
			} else {
				// 并发时用主键冲突控制
				try {
					mqConsumerLog = new TradeMqConsumerLog();
					mqConsumerLog.setGroupName(groupName);
					mqConsumerLog.setMsgKeys(keys);
					mqConsumerLog.setMsgTag(tags);
					mqConsumerLog.setConsumerTimes(0);
					mqConsumerLog.setMsgId(msgId);
					mqConsumerLog.setMsgBody(body);
					mqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode());
					this.tradeMqConsumerLogMapper.insertSelective(mqConsumerLog);
				} catch (Exception e) {
					LOGGER.warn("主键冲突，说明有订阅者正在处理，稍后再试");
					return false;
				}
			}
			CancelOrderMQ cancelOrderMQ = JSON.parseObject(body, CancelOrderMQ.class);
			AddGoodsNumberReq addGoodsNumberReq = new AddGoodsNumberReq();
			addGoodsNumberReq.setGoodsId(cancelOrderMQ.getGoodsId());
			addGoodsNumberReq.setGoodsNumber(cancelOrderMQ.getGoodsNumber());
			addGoodsNumberReq.setOrderId(cancelOrderMQ.getOrderId());
			tradeGoodsService.addGoodsNumber(addGoodsNumberReq);

			// 更新处理成功
			TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
			updateMqConsumerLog.setId(mqConsumerLog.getId());
			updateMqConsumerLog.setGroupName(groupName);
			updateMqConsumerLog.setMsgKeys(keys);
			updateMqConsumerLog.setMsgTag(tags);
			updateMqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() + 1);
			updateMqConsumerLog.setMsgId(msgId);
			updateMqConsumerLog.setMsgBody(body);
			updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.SUCCESS.getStatusCode());
			this.tradeMqConsumerLogMapper.updateByPrimaryKeySelective(updateMqConsumerLog);
			return true;
		} catch (Exception e) {
			TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
			updateMqConsumerLog.setGroupName(mqConsumerLog.getGroupName());
			updateMqConsumerLog.setMsgKeys(mqConsumerLog.getMsgKeys());
			updateMqConsumerLog.setMsgTag(mqConsumerLog.getMsgTag());
			updateMqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() + 1);
			updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.FAIL.getStatusCode());
			this.tradeMqConsumerLogMapper.updateByPrimaryKeySelective(updateMqConsumerLog);
			return false;
		}

	}
}
